#ifndef __MQ_MESSAGE_
#define __MQ_MESSAGE_

#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_tools.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <string>
#include <list>
#include <mutex>
#include <unordered_map>
#include <memory>

namespace MQ
{
    using MessagePtr = std::shared_ptr<MQ::Message>;

    class MessageMapper
    {
    public:
        MessageMapper(const std::string &basedir, const std::string &qname)
            : _qname(qname)
        {
            std::string _basedir = basedir;
            if (_basedir.back() != '/')
                _basedir += "/";

            _datafile = _basedir + qname + ".mqd";
            _tmpfile = _basedir + qname + ".mqd.tmp";

            if (FileHelper(_basedir).exists() == false) // 防止初始化的时候被覆盖
                assert(FileHelper::createDirectory(_basedir));

            createMsgFile();
        }

        void removeMsgFile()
        {
            FileHelper::removeFile(_datafile);
            FileHelper::removeFile(_tmpfile);
        }

        bool insert(const MessagePtr &msp)
        {
            return insert(_datafile, msp);
        }

        bool remove(MessagePtr &msp)
        {
            // 更改有效标志位为0
            msp->mutable_payload()->set_valid("0");
            // 序列化
            std::string body = msp->payload().SerializeAsString();
            if (body.size() != msp->length())
            {
                LOG(ERROR, "序列化后的数据长度与原数据长度不一致!!!\n");
                return false;
            }
            // 重新写入到文件
            FileHelper fp(_datafile);
            bool ret = fp.write(body.c_str(), msp->offset(), body.size());
            if (ret == false)
            {
                LOG(ERROR, "向队列文件的删除数据写入失败!!!\n");
                return false;
            }
            return true;
        }

        std::list<MessagePtr> gc()
        {
            // 获取有效数据
            std::list<MessagePtr> lmsp;
            bool ret = load(lmsp);
            if (ret == false)
            {
                LOG(ERROR, "有效数据链表加载失败!!!\n");
                return std::list<MessagePtr>();
            }
            // 有效数据写入到临时文件
            FileHelper::createFile(_tmpfile);
            for (auto &msg : lmsp)
            {
                LOG(INFO, "向临时文件写入数据ing...\n");
                ret = insert(_tmpfile, msg);
                if (ret == false)
                {
                    LOG(ERROR, "向临时文件写入数据失败!!!\n");
                    return std::list<MessagePtr>();
                }
            }

            // 删除源文件
            ret = FileHelper::removeFile(_datafile);
            if (ret == false)
            {
                LOG(ERROR, "删除源文件失败!!!\n");
                return lmsp;
            }
            // 修改临时文件为源文件

            ret = FileHelper(_tmpfile).rename(_datafile);
            if (ret == false)
            {
                LOG(ERROR, "修改文件名称失败!!!\n");
                return lmsp;
            }
            // 返回有效数据链表
            return lmsp;
        }

    private:
        void createMsgFile()
        {
            if (FileHelper(_datafile).exists())
                return;

            bool ret = FileHelper::createFile(_datafile);
            if (ret == false)
            {
                LOG(ERROR, "创建队列数据文件失败!!!\n");
                return;
            }
        }

        bool insert(const std::string &_datafile, const MessagePtr &msp)
        {
            // 序列化数据
            std::string body = msp->payload().SerializeAsString();
            // 读取文件
            FileHelper fp(_datafile);
            size_t fsize = fp.size();
            size_t msize = body.size();
            // 写入策略 长度+内容
            bool ret = fp.write(reinterpret_cast<const char *>(&msize), fsize, sizeof(size_t));
            if (ret == false)
            {
                LOG(ERROR, "写入文件数据长度失败!!!\n");
                return false;
            }
            ret = fp.write(body.c_str(), fsize + sizeof(size_t), msize);
            if (ret == false)
            {
                LOG(ERROR, "写入文件数据内容失败!!!\n");
                return false;
            }
            // 更新
            msp->set_offset(fsize + sizeof(size_t));
            msp->set_length(body.size());
            return true;
        }

        // 加载所有的有效数据
        bool load(std::list<MessagePtr> &msp)
        {
            // 消息的基本格式 4字节长度 | 有效数据 | 4字节长度 | 有效数据
            FileHelper fp(_datafile);
            size_t fsize = fp.size();
            size_t offset = 0, msg_size = 0;
            bool ret = false;
            while (offset < fsize)
            {
                // 长度
                ret = fp.read(reinterpret_cast<char *>(&msg_size), offset, sizeof(size_t));
                if (ret == false)
                {
                    LOG(ERROR, "加载消息长度失败!!!\n");
                    return false;
                }
                offset += sizeof(size_t);
                // body
                std::string msg_body;
                msg_body.resize(msg_size);
                ret = fp.read(&msg_body[0], offset, msg_size);
                if (ret == false)
                {
                    LOG(ERROR, "加载消息数据失败!!!\n");
                    return false;
                }
                offset += msg_size;

                // 插入到list中
                MessagePtr mes = std::make_shared<MQ::Message>();
                // 需要判断是否有效
                mes->mutable_payload()->ParseFromString(msg_body); // 修改使用mutable_*
                if (mes->payload().valid() == "0")
                {
                    LOG(ERROR, "加载到无用数据!!!内容为:%s\n", mes->payload().body());
                    continue;
                }

                msp.push_back(mes);
            }
            return true;
        }

    private:
        std::string _qname;
        std::string _datafile;
        std::string _tmpfile;
    };

    class QueueMessage
    {
    public:
        using ptr = std::shared_ptr<QueueMessage>;

        QueueMessage(const std::string &basedir, const std::string &qname)
            : _mapper(basedir, qname), _total_count(0), _valid_count(0), _qname(qname)
        {
        }

        bool insert(const MQ::BasicProperties *basic, const std::string &body, bool durable)
        {
            // 构建消息对象
            MessagePtr msp = std::make_shared<MQ::Message>();
            msp->mutable_payload()->set_body(body);
            if (basic)
            {
                MQ::DeliveryMode mode = durable ? basic->delivery_mode() : MQ::DeliveryMode::UNDURABLE;
                msp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msp->mutable_payload()->mutable_properties()->set_id(basic->id());
                msp->mutable_payload()->mutable_properties()->set_routing_key(basic->routing_key());
            }
            else
            {
                MQ::DeliveryMode mode = durable ? MQ::DeliveryMode::DURABLE : MQ::DeliveryMode::UNDURABLE;
                msp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msp->mutable_payload()->mutable_properties()->set_id(UuidHelper::uuid());
                msp->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            // 是否需要持久化 -> CURD +mutex
            std::unique_lock<std::mutex> lock(_mutex);
            if (msp->payload().properties().delivery_mode() == MQ::DeliveryMode::DURABLE)
            {
                // 进行持久化存储
                msp->mutable_payload()->set_valid("1");
                bool ret = _mapper.insert(msp);
                if (ret == false)
                {
                    LOG(ERROR, "消息持久化失败!!!\n");
                    return false;
                }
                _valid_count += 1;
                // 持久化插入数据
                _durable_msgs.insert(std::make_pair(msp->payload().properties().id(), msp));
            }
            _total_count += 1;
            // 内存管理
            _msgs.push_back(msp);
            return true;
        }

        // ack之后判断是否需要gc
        bool remove(const std::string &msg_id)
        {
            // 从ack队列中找)
            auto it = _waitack_msgs.find(msg_id);
            if (it == _waitack_msgs.end())
            {
                LOG(WARNING, "remove不存在该条消息!!!\n");
                return false;
            }
            // 根据消息的持久化模式
            if (it->second->payload().properties().delivery_mode() == MQ::DeliveryMode::DURABLE)
            {
                // 删除持久化信息 mapper durable_map
                _mapper.remove(it->second);
                _durable_msgs.erase(msg_id);
                _valid_count -= 1;
                gc();
            }
            // 删除内存信息
            _waitack_msgs.erase(msg_id);
            return true;
        }

        MessagePtr front()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            if (_msgs.empty())
            {
                LOG(WARNING, "消息队列为空!!!\n");
                return MessagePtr();
            }

            MessagePtr msp = _msgs.front();
            _msgs.pop_front();
            _waitack_msgs.insert(std::make_pair(msp->payload().properties().id(), msp));
            return msp;
        }

        bool recovery()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _msgs = _mapper.gc();
            if (_msgs.empty())
                return true;
            for (auto &msg : _msgs)
            {
                _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
            }
            _valid_count = _total_count = _msgs.size();
            return true;
        }

        size_t valid_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msgs.size();
        }

        size_t total_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _total_count;
        }

        size_t waitack_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _waitack_msgs.size();
        }

        size_t durable_count()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _durable_msgs.size();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeMsgFile();
            _msgs.clear();
            _durable_msgs.clear();
            _waitack_msgs.clear();
            _total_count = _valid_count = 0;
        }

    private:
        bool GcCheck()
        {
            if (_total_count >= 2000 && (_valid_count * 100 / _total_count) < 50)
                return true;
            return false;
        }

        void gc()
        {
            if (GcCheck() == false)
                return;

            std::list<MessagePtr> msp = _mapper.gc();
            for (auto &msg : msp)
            {
                auto it = _durable_msgs.find(msg->payload().properties().id());
                if (it == _durable_msgs.end())
                {
                    LOG(WARNING, "有一条持久化消息未进行内存管理!!!\n");
                    _msgs.push_back(msg);
                    _durable_msgs.insert(std::make_pair(msg->payload().properties().id(), msg));
                }
                // 更新存储位置
                it->second->set_offset(msg->offset());
                it->second->set_offset(msg->length());
            }
            // 更新消息数量数据
            _valid_count = _total_count = _msgs.size();
        }

    private:
        std::mutex _mutex;
        std::string _qname;
        size_t _total_count;
        size_t _valid_count;
        MessageMapper _mapper;
        std::list<MessagePtr> _msgs;                               // 有效消息列表
        std::unordered_map<std::string, MessagePtr> _durable_msgs; // 持久化消息
        std::unordered_map<std::string, MessagePtr> _waitack_msgs; // 待确认消息
    };

    class MessageManager
    {
    public:
        using ptr = std::shared_ptr<MessageManager>;

        MessageManager(const std::string &basedir)
            : _basedir(basedir)
        {
        }

        void InitQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmsp;
            {
                auto it = _queue_msg.find(qname);
                if (it != _queue_msg.end())
                {
                    LOG(INFO, "Init已经存在该消息队列!!!\n");
                    return;
                }
                qmsp = std::make_shared<QueueMessage>(_basedir, qname);
                _queue_msg.insert(std::make_pair(qname, qmsp));
            }
            qmsp->recovery();
        }

        void DestoryQueueMessage(const std::string &qname)
        {
            QueueMessage::ptr qmsp;
            {
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(INFO, "DeStory不存在该消息队列!!!\n");
                    return;
                }
                qmsp = it->second;
                _queue_msg.erase(it);
            }
            qmsp->clear();
        }

        bool insert(const std::string &qname, MQ::BasicProperties *basic, const std::string &body, bool durable)
        {
            QueueMessage::ptr qmsp;
            {
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(INFO, "insert不存在该消息队列%s!!!\n", qname.c_str());
                    return false;
                }
                qmsp = it->second;
            }
            return qmsp->insert(basic, body, durable);
        }

        void ack(const std::string &qname, const std::string &msg_id)
        {
            QueueMessage::ptr qmsp;
            {
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(INFO, "ack不存在该消息队列!!!\n");
                    return;
                }
                qmsp = it->second;
            }
            qmsp->remove(msg_id);
        }

        MessagePtr front(const std::string &qname)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _queue_msg.find(qname);
            if (it == _queue_msg.end())
            {
                LOG(WARNING, "front不存在该消息队列!!!\n");
                return MessagePtr();
            }
            return it->second->front();
        }

        size_t total_count(const std::string &qname)
        {
            QueueMessage::ptr qmsp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(WARNING, "total_count不存在该消息队列!!!\n");
                    return 0;
                }
                qmsp = it->second;
            }
            return qmsp->total_count();
        }

        size_t valid_count(const std::string &qname)
        {
            QueueMessage::ptr qmsp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(WARNING, "valid_count不存在该消息队列!!!\n");
                    return 0;
                }
                qmsp = it->second;
            }
            return qmsp->valid_count();
        }

        size_t durable_count(const std::string &qname)
        {
            QueueMessage::ptr qmsp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(WARNING, "durable_count不存在该消息队列!!!\n");
                    return 0;
                }
                qmsp = it->second;
            }
            return qmsp->durable_count();
        }

        size_t waitack_count(const std::string &qname)
        {
            QueueMessage::ptr qmsp;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _queue_msg.find(qname);
                if (it == _queue_msg.end())
                {
                    LOG(WARNING, "waitack_count不存在该消息队列!!!\n");
                    return 0;
                }
                qmsp = it->second;
            }
            return qmsp->waitack_count();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            for (auto &qmsg : _queue_msg)
            {
                qmsg.second->clear();
            }
        }

    private:
        std::mutex _mutex;
        std::string _basedir;
        std::unordered_map<std::string, QueueMessage::ptr> _queue_msg;
    };
}
#endif